package bigdata.backend.monitor.binlog.service;

import bigdata.backend.monitor.utils.KafkaProgressUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * Kafka 消费进度监控服务 (针对 binlog 项目 - Backend 模块)
 * 负责查询不同 Topic 对应的不同消费者组的进度
 */
@Service
public class KafkaMonitorService {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    // 从配置文件获取不同 Topic 对应的消费者组 ID
    @Value("${monitor.binlog.consumer-group-id:binlog-monitor-group}")
    private String binlogConsumerGroupId;

    @Value("${monitor.violation.consumer-group-id:monitor-violation-events-group}")
    private String violationConsumerGroupId;

    @Value("${monitor.initial-topics:binlog-monitor}")
    private String initialTopic;

    @Value("${monitor.violation-topic:monitor-violation-events}")
    private String violationTopic;

    /**
     * 查询 binlog 监控 Topic (initial-topics) 的消费进度
     * 该 Topic 由 binlog-monitor-group 消费
     * @return TopicProgressInfo 对象
     */
    public KafkaProgressUtils.TopicProgressInfo getBinlogTopicProgress() {
        System.out.println("Querying progress for initial topic: " + initialTopic + " with group: " + binlogConsumerGroupId);
        return KafkaProgressUtils.getTopicProgress(bootstrapServers, binlogConsumerGroupId, initialTopic);
    }

    /**
     * 查询违规事件 Topic (violation-topic) 的消费进度
     * 该 Topic 由 monitor-violation-events-group 消费
     * @return TopicProgressInfo 对象
     */
    public KafkaProgressUtils.TopicProgressInfo getViolationTopicProgress() {
        System.out.println("Querying progress for violation topic: " + violationTopic + " with group: " + violationConsumerGroupId);
        return KafkaProgressUtils.getTopicProgress(bootstrapServers, violationConsumerGroupId, violationTopic);
    }

    /**
     * 查询指定 Topic 的消费进度 (通用方法)
     * @param topicName 要查询的 Topic 名称
     * @param consumerGroupId 该 Topic 对应的消费者组 ID
     * @return TopicProgressInfo 对象
     */
    public KafkaProgressUtils.TopicProgressInfo getTopicProgress(String topicName, String consumerGroupId) {
        System.out.println("Querying progress for topic: " + topicName + " with group: " + consumerGroupId);
        return KafkaProgressUtils.getTopicProgress(bootstrapServers, consumerGroupId, topicName);
    }

    // Getters for configuration values (optional, if needed elsewhere)
    public String getBootstrapServers() {
        return bootstrapServers;
    }

    public String getBinlogConsumerGroupId() {
        return binlogConsumerGroupId;
    }

    public String getViolationConsumerGroupId() {
        return violationConsumerGroupId;
    }

    public String getInitialTopic() {
        return initialTopic;
    }

    public String getViolationTopic() {
        return violationTopic;
    }
}